package com.neusoft.dibuy.partitioner;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

public class PhoneDataMapper extends Mapper<LongWritable, Text, Text, FlowBean> {
    private final Text key_out = new Text();

    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, FlowBean>.Context context) throws IOException, InterruptedException {
        String[] words = value.toString().split("\t");
        if (words.length == 7) {
            String phone = words[1];
            if (phone.length() != 11) {
                return;
            }
            key_out.set(phone);
            FlowBean flowBean = new FlowBean();
            flowBean.setPhone(phone);
            long upFlow = Long.parseLong(words[4]);
            long downFlow = Long.parseLong(words[5]);
            flowBean.setUpFlow(upFlow);
            flowBean.setDownFlow(downFlow);
            flowBean.setSumFlow();
            System.out.println(key_out.toString() + "----> " + flowBean.toString());
            context.write(key_out, flowBean);
        } else {
            return;
        }
    }
}
